package netty.mq;

import com.rabbitmq.client.*;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

public class RabbitMQManager {
    private Connection connection;
    private Channel channel;
    private final String host;
    private final int port;
    private final String username;
    private final String password;
    private final String virtualHost;
    private final ExecutorService executor;

    public RabbitMQManager(String host, int port, String username,
                           String password, String virtualHost) {
        this.host = host;
        this.port = port;
        this.username = username;
        this.password = password;
        this.virtualHost = virtualHost;
        this.executor = Executors.newFixedThreadPool(
                4,
                new DefaultThreadFactory("rabbitmq-worker")
        );
    }

    // 初始化连接（需在Netty服务启动时调用）
    public void start() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(username);
        factory.setPassword(password);
        factory.setVirtualHost(virtualHost);
        factory.setAutomaticRecoveryEnabled(true); // 自动重连
        factory.setNetworkRecoveryInterval(5000);  // 重连间隔5秒

        this.connection = factory.newConnection(executor);
        this.channel = connection.createChannel();

        // 声明死信队列（示例）
        channel.queueDeclare("dead.letter.queue", true, false, false, null);
    }

    // 发送消息（异步）
    public void publish(String exchange, String routingKey, byte[] body) {
        try {
            channel.basicPublish(
                    exchange,
                    routingKey,
                    new AMQP.BasicProperties.Builder()
                            .contentType("application/json")
                            .deliveryMode(2) // 持久化消息
                            .build(),
                    body
            );
        } catch (IOException e) {
            System.err.println("消息发送失败: " + e.getMessage());
        }
    }

    public void consume(String queueName) throws IOException {
        // 1. 声明队列（如果不存在）
        channel.queueDeclare(queueName, true, false, false, null);

        // 2. 创建消费者（4.1.0版本用法）
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(
                    String consumerTag,
                    Envelope envelope,
                    AMQP.BasicProperties properties,
                    byte[] body
            ) throws IOException {
                // 处理消息
                String message = new String(body, "UTF-8");
                System.out.println("收到消息: " + message);

                // 手动ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };

        // 3. 开始消费（关闭自动ACK）
        channel.basicConsume(queueName, false, consumer);
    }

    // 关闭连接（需在Netty服务关闭时调用）
    public void shutdown() {
        try {
            if (channel != null && channel.isOpen()) {
                channel.close();
            }
            if (connection != null && connection.isOpen()) {
                connection.close();
            }
            executor.shutdown();
        } catch (IOException | TimeoutException e) {
            System.err.println("关闭RabbitMQ连接异常: " + e.getMessage());
        }
    }
}